package com.star.batchhandler.demo;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import javax.sql.DataSource;

public class SQLCommandScheduler implements ISQLCommandScheduler{
	public final static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(SQLBatchOperation.batch_stmt_size);
	// public static long executeTime = 0;
	// private long sycTime = 0;
	private DataSource ds;
	private List<Connection> conns = new ArrayList<Connection>();
	List<Statement> stmts = new ArrayList<Statement>();
	private volatile int currentJobs;
	// private volatile int readyJobs;
	// private Object readyLock = new Object();
	private int[] batchResults;
	// private ConcurrentHashMap<Command, Statement> stmts = new
	// ConcurrentHashMap<Command, Statement>();
	// private List<Thread> ts = new ArrayList<Thread>();
	// private ConcurrentHashMap<Command, Integer> batchResults = new
	// ConcurrentHashMap<Command, Integer>();
//	private Object[] locks = new Object[SQLBatchOperation.batch_stmt_size];
//	private int batch_size_stmt = SQLBatchOperation.con_size/SQLBatchOperation.batch_stmt_size + 1;
	private ArrayList<Command> commands = new ArrayList<Command>();
	private ArrayList<Command> backup_commands = new ArrayList<Command>();
	private boolean isFinished = false;

	private Thread addJobThread;

	public SQLCommandScheduler(DataSource ds, Thread addJobThread) throws SQLException {
		this.ds = ds;
		this.addJobThread = addJobThread;
		for (int i = 0;i<SQLBatchOperation.batch_stmt_size;i++) {
			try {
				Connection conn = ds.getConnection();
				conn.setAutoCommit(false);
				conns.add(conn);
			} catch (SQLException e) {
				e.printStackTrace();
				throw new IllegalStateException("",e);
			}
		}
	}

	@Override
	public void run() {

	}

	// public void addCurrentThread(Thread t) {
	// this.ts.add(t);
	// }

	public void addCommand(Command cmd) throws Exception {
		// System.out.println("add 1");
		// synchronized (readyLock) {
		// // System.out.println("add 2");
		// readyJobs++;
		// if (isReady(readyJobs)) {
		// commands.clear();
		// readyJobs = 0;
		// readyLock.notifyAll();
		// } else {
		// readyLock.wait();
		// }
		// }
		// System.out.println("add 3");
//		System.out.println("add00 "+cmd);
			commands.add(cmd);
			synchronized (this) {
//			System.out.println("add01 "+cmd);
			currentJobs++;
			// AbstractJob.consumTime += (System.nanoTime() - begin);
			// System.out.println("add 5.currentJobs:"+currentJobs);
			if (currentJobs == SQLBatchOperation.jobsInBatch) {
//				System.out.println("add02-batch "+cmd);
				batch();
			} else {
//				System.out.println("add02-wait "+cmd);
				this.wait();
			}
		}
	}

	public void batch() throws Exception {
		// for (Command cmd : stmts.keySet()) {
		// Statement stmt = stmts.remove(cmd);
		// batchResults.put(cmd, stmt.executeBatch());
		// stmt.close();
		// }
		// System.out.println("add 7.");
		// long begin = System.nanoTime();
//		if (commands.get(0).getParams() == null) {
//			for(int i = 0;i<SQLBatchOperation.batch_stmt_size;i++) {
//				stmts.add(conn.createStatement());
//			}
//			
//		} else {
//			for(int i = 0;i<SQLBatchOperation.batch_stmt_size;i++) {
//				stmts.add(conn.prepareStatement(commands.get(0).getSql()));
//			}
//			
//		}
		final Status status = new Status();
//		System.out.println("add03-inbatch ");
		for (int i = 0;i<SQLBatchOperation.batch_stmt_size;i++) {
			scheduler.execute(new Batcher(i,ds,status));
		}
		
		while(true){
			if(status.isFinished){
//				for (Connection conn : conns) {
//					conn.close();
//				}
//				conns.clear();
				recover();
				this.notifyAll();
//				System.out.println("add03-outbatch ");
				break;
			}
			synchronized (status) {
				status.wait();
			}
		}
	}

	class Status{
		volatile int count = 0;
		volatile boolean isFinished = false;
		
		public int selfAdd(){
			return ++count;
		}
		public void finish(){
			this.isFinished = true;
		}
	}
	private void recover() {
		backup_commands.clear();
		backup_commands.addAll(commands);
		commands.clear();
		currentJobs = 0;
//		stmts.clear();
	}

	private boolean isReady(int jobs) {
		return jobs == SQLBatchOperation.jobsInBatch;
	}

	public int getBatchResult(Command cmd) {
//		return batchResults[backup_commands.indexOf(cmd)];
		return 2;
	}

	public boolean isFinished() {
		return isFinished;
	}

	public void setFinished() {
		synchronized (addJobThread) {
			addJobThread.notify();
		}
	}
	
	
	class Batcher implements Runnable{
		private Status status;
		private int index;
		private Connection conn;
		public Batcher(int index,DataSource ds,Status status) {
			this.index = index;
			conn = conns.get(index);
			this.status = status;
		}
		@Override
		public void run() {
			Statement stmt = null;
			try{
			if (commands.get(0).getParams() == null) {
				stmt = conn.createStatement();
				for (int i = index; i<commands.size(); i += SQLBatchOperation.batch_stmt_size ) {
					stmt.addBatch(commands.get(i).getSql());
				}
			} else {
				PreparedStatement ps = (PreparedStatement) conn.prepareStatement(commands.get(0).getSql());
				stmt = ps;
				for (int i = index; i<commands.size(); i += SQLBatchOperation.batch_stmt_size ) {
					try{
						ps.setLong(1, (Long) commands.get(i).getParams()[0]);
						ps.addBatch();
					}catch(Exception e){
						e.printStackTrace();
					}
					// stmts.put(cmd, ps);
				}
			}
			batchResults = stmt.executeBatch();
			// System.out.println("add 8.");
			conn.commit();
			// System.out.println("add 9.");
			stmt.close();
//			status.selfAdd();
			if(index == SQLBatchOperation.batch_stmt_size-1){
				status.finish();
			}
			synchronized (status) {
				status.notifyAll();
			}
			}catch(Exception e){
				e.printStackTrace();
				throw new IllegalStateException("", e);
			}
		}
	}
	public volatile Integer jobs = 0;
	public Integer selfAddJob(){
		return ++jobs;
	}
	public void initJobs(){
		jobs = 0;
	}
	public Integer selfDelJob(){
		return --jobs;
	}
	public Integer getJobs(){
		return jobs;
	}

}
